package com.grf.kafka.java.interceptor;

import com.grf.kafka.java.KafkaConstant;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * @author grf
 * @apiNote 生产者-拦截器
 * @since 2024/10/27
 */
@Slf4j
public class InterceptorProduct {
    @SneakyThrows
    public static void main(String[] args) {
        //PART1:设置发送者相关属性
        Properties props = new Properties();
        // 此处配置的kafka的端口
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstant.BOOTSTRAP_SERVERS);
        // 配置key的序列化类
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.grf.kafka.java.interceptor.MyProductInterceptor");
        org.apache.kafka.clients.producer.Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i=0; i<10; i++){
            // part2: 构建消息
            ProducerRecord<String, String> record = new ProducerRecord<>(KafkaConstant.TOPIC,Integer.toString(i), "MyProducer "+i);
            // 同步发送，获取服务端应答消息前，会阻塞当前线程
            RecordMetadata metadata = producer.send(record).get();
            String topic = metadata.topic();
            int partition = metadata.partition();
            long offset = metadata.offset();
            String message = metadata.toString();
            log.info("topic is {}; partition is {}; offset is {}; message is {}", topic, partition, offset, message);
        }
        producer.close();
    }
}
